package link.jfire.jnet.server.transfer.CompletionHandler;

import java.nio.channels.AsynchronousCloseException;
import java.nio.channels.AsynchronousSocketChannel;
import java.nio.channels.ClosedChannelException;
import java.nio.channels.CompletionHandler;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.ThreadFactory;
import link.jfire.baseutil.simplelog.ConsoleLogFactory;
import link.jfire.baseutil.simplelog.Logger;
import link.jfire.jnet.server.storage.EntryAction;
import link.jfire.jnet.server.storage.ringarray.ComplexMultRingArray;
import link.jfire.jnet.server.storage.ringarray.RingArray;
import link.jfire.jnet.server.storage.ringarray.SimpleMultRingArray;
import link.jfire.jnet.server.transfer.server.AioServer;
import link.jfire.jnet.server.transfer.server.ChannelInitListener;
import link.jfire.jnet.server.transfer.server.ServerChannelInfo;
import link.jfire.jnet.server.transfer.server.ServerConfig;

public class AcceptHandler implements CompletionHandler<AsynchronousSocketChannel, Object>
{
	private AioServer				aioServer;
	private Logger					logger				= ConsoleLogFactory.getLogger();
	private ServerConfig			config;
	private ChannelReadHandler		channelReadHandler;
	private ChannelWriteHandler		channelWriteHandler	= new ChannelWriteHandler();
	private ChannelInitListener[]	initListeners;
	private RingArray				ringArray;
	private ExecutorService			threadPool;
									
	public AcceptHandler(AioServer aioServer, ServerConfig serverConfig)
	{
		config = serverConfig;
		channelReadHandler = new ChannelReadHandler(serverConfig.getFrameDecodec(), serverConfig.getCloseListeners());
		this.initListeners = serverConfig.getInitListeners();
		this.aioServer = aioServer;
		EntryAction[] actions = new EntryAction[serverConfig.getHandlerThreadSize()];
		switch (serverConfig.getRingArrayTpe())
		{
			case ServerConfig.SimpleMult:
				ringArray = new SimpleMultRingArray(serverConfig.getRingArraySize(), serverConfig.getWaitStrategy(), actions);
				break;
			case ServerConfig.ComplexMult:
				ringArray = new ComplexMultRingArray(serverConfig.getRingArraySize(), serverConfig.getWaitStrategy(), actions);
		}
		for (int i = 0; i < actions.length; i++)
		{
			actions[i] = new EntryAction(ringArray);
		}
		threadPool = Executors.newFixedThreadPool(actions.length, new ThreadFactory() {
			int i = 0;
			
			@Override
			public Thread newThread(Runnable r)
			{
				return new Thread(r, "任务线程-" + (i++));
			}
		});
		for (int i = 0; i < actions.length; i++)
		{
			threadPool.execute(actions[i]);
		}
	}
	
	public void stop()
	{
		threadPool.shutdown();
		ringArray.stop();
	}
	
	@Override
	public void completed(AsynchronousSocketChannel socketChannel, Object attachment)
	{
		try
		{
			ServerChannelInfo channelInfo = new ServerChannelInfo(socketChannel, channelReadHandler, channelWriteHandler, ringArray);
			channelInfo.setReadTimeout(config.getReadTiemout());
			channelInfo.setWaitTimeout(config.getWaitTimeout());
			if (initListeners != null)
			{
				for (ChannelInitListener each : initListeners)
				{
					each.channelInit(channelInfo);
				}
			}
			logger.debug("开启一个新通道{}", channelInfo.getAddress());
			channelInfo.startReadWait();
			aioServer.getServerSocketChannel().accept(null, this);
		}
		catch (Exception e)
		{
			logger.error("注册异常", e);
		}
	}
	
	@Override
	public void failed(Throwable exc, Object attachment)
	{
		if (exc instanceof AsynchronousCloseException)
		{
			logger.info("服务端监听链接被关闭");
		}
		else if (exc instanceof ClosedChannelException)
		{
			logger.info("服务端监听链接被关闭");
		}
		else
		{
			logger.error("链接异常关闭", exc);
		}
		Thread.currentThread().interrupt();
	}
	
}
